package com.burges.net.dataSet.api.transform

import org.apache.flink.api.scala.{DataSet, ExecutionEnvironment}

/**
  * 创建人    BurgessLee 
  * 创建时间   2020/2/11 
  * 描述      outerJoin算子的实例代码
  */
object OuterJoinOperatorDemo {

	def main(args: Array[String]): Unit = {
		var environment: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
		val caseClassPersonDs1: DataSet[Person] = environment.fromElements(Person(1, "a"),Person(1, "a"))
		val caseClassPersonDs2: DataSet[(Double, Int)] = environment.fromElements((12.3,1), (22.3,3))

		// 左外连接两个数据集，按照相同的key进行关联，如果右边数据集没有数据会填充空值
		caseClassPersonDs1.leftOuterJoin(caseClassPersonDs2).where("id").equalTo(1)
		// 右外连接两个数据集，按照系统的key进行关联，如果左边数据集中没有数据则填充空值
		caseClassPersonDs1.rightOuterJoin(caseClassPersonDs2).where("id").equalTo(1)

		// 第二种使用方式
		var caseClassRs2 = caseClassPersonDs1.leftOuterJoin(caseClassPersonDs2).where("age").equalTo(1){
			(left, right) => (left.age, left.name, right._1 +1)
		}

		//大数据集也会有相应的算法提示操作
		/**
		  * 当然也有其他操作，操作类似Join的使用方法示例代码
		  */
		// 将第一个数据集广播出去，转换成HashTable存储，适用于第一个数据集非常小的情况
//		caseClassPersonDs1.leftOuterJoin(caseClassPersonDs2, BROADCAST_HASH_FIRST).where("id").equalTo(1)
		// 将两个数据集重新分区，并将第一个数据集转换成HashTable存储，该策略适用于第一个数据集比第二个数据集小，但两个数据集相对比较大的情况
//		caseClassPersonDs1.leftOuterJoin(caseClassPersonDs2, REPARTITION_HASH_FIRST).where("id").equalTo(1)

	}

}

